package com.mxq.eclipse.mqtt.demo.subscribe.service.impl;


import com.mxq.eclipse.mqtt.demo.subscribe.MqttSubscribeDto;
import com.mxq.eclipse.mqtt.demo.subscribe.SubscribeConnect;
import com.mxq.eclipse.mqtt.demo.subscribe.service.MqttSubscribeService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * @author xiaoqiang
 * @date 2020-05-09 16:09
 */
@Slf4j
@Service
public class MqttSubscribeServiceImpl implements MqttSubscribeService {

    @Autowired
    private SubscribeConnect subscribeConnect;

    @Override
    public Boolean subscribe(MqttSubscribeDto mqttSubscribeDto) {
        log.info("MQ===subscribe=== 入参:mqttSubscribeDto:{}", mqttSubscribeDto);
        MqttClient mqttClient = subscribeConnect.getMqttClient();
        // 判定是否需要重新连接
        if (mqttClient==null || !mqttClient.isConnected()) {
            mqttClient = subscribeConnect.getConn(mqttSubscribeDto);
        }
        try {
            // 订阅消息
            //int[] qos = {mqttSubscribeDto.getQos() == null ? 1 : mqttSubscribeDto.getQos()};
            int[] qos = new int[mqttSubscribeDto.getTopics().length];
            for (int i=0; i<qos.length; i++) {
                qos[i] = mqttSubscribeDto.getQos() == null ? 1 : mqttSubscribeDto.getQos();
            }
            mqttClient.subscribe(mqttSubscribeDto.getTopics(), qos);
        } catch (MqttException e) {
            e.printStackTrace();
            log.info("主题订阅失败！errorMessage:{}", e.getMessage());
            return false;
        }
        return true;
    }
}
